package cn.bby.query.server.domain.share.util;

import cn.bby.jtk.base.constant.CommonErrorCode;
import cn.bby.jtk.base.exception.ServiceException;
import cn.hutool.core.util.SerializeUtil;
import cn.hutool.db.Db;
import cn.hutool.db.DbUtil;
import cn.hutool.db.Entity;
import cn.hutool.db.ds.simple.SimpleDataSource;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.sql.SQLException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author wangbenyin
 * @version 1.0
 * @date 2025/07/02
 * @since 17
 */
@Component
public class CrossDbUtil {
    public static CrossDbUtil instance;
    private static final Map<String, Db> DB_CACHE = new ConcurrentHashMap<>();

    @Value("${debezium.database.hostname}")
    private String hostname;
    @Value("${debezium.database.port}")
    private String port;
    @Value("${debezium.database.user}")
    private String user;
    @Value("${debezium.database.password}")
    private String password;

    public CrossDbUtil() {
        CrossDbUtil.instance = this;
    }

    public Db getDB(String dbname) {
        return DB_CACHE.computeIfAbsent(dbname, k -> {
            String dbUrl = "jdbc:postgresql://%s:%s/%s".formatted(hostname, port, dbname);
            SimpleDataSource dataSource = new SimpleDataSource(dbUrl, user, password);
            dataSource.setDriver("org.postgresql.Driver");
            return DbUtil.use(dataSource);
        });
    }

    public Entity queryOne(String dbname, String sql, Object... params) {
        try {
            return getDB(dbname).queryOne(sql, params);
        } catch (SQLException e) {
            throw new ServiceException(CommonErrorCode.SYSTEM_ERROR, e);
        }
    }
    public void execute(String dbname, String sql, Object... params) {
        try {
            getDB(dbname).execute(sql, params);
        } catch (SQLException e) {
            throw new ServiceException(CommonErrorCode.SYSTEM_ERROR, e);
        }
    }

    public void updateOffset(String topic, Map<byte[], byte[]> data) {
        byte[] bytes = SerializeUtil.serialize(data);
        execute("audit_log", "update al_offset set data = ? where topic = ?", bytes, topic);
    }

    public Map<byte[], byte[]> queryOffset(String topic) {
        Entity entity = queryOne("audit_log", "select data from al_offset where topic = ?", topic);
        return entity == null ? null : SerializeUtil.deserialize(entity.getBytes("data"));
    }

    public void createOffset(String topic) {
        execute("audit_log", "insert into al_offset (topic,data) values (?, ?)", topic, SerializeUtil.serialize(new HashMap<byte[], byte[]>()));
    }
}
